package thread;

import org.apache.commons.compress.compressors.z.ZCompressorInputStream;
import org.apache.commons.lang.StringUtils;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.CountDownLatch;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

/**
 * Created by fanzhenyu02 on 2020/6/24.
 */
public class MultiThreadRunner {

    private static final int DEFAULT_BUFFER_SIZE = 1048576;
    private static final int DEFAULT_SPLIT_FILE_SIZE = 16;

    public static void main(String[] args) {
        MultiThreadRunner an = new MultiThreadRunner();
        System.out.println(an.convertFileFormat("", "/Users/vonmax/Downloads/merge/tempRawBankFile.z", "final.txt"));
    }


    /**
     * 处理银行原始账单,每条流水记录最后增加两个字段: 银行名称+卡类型标识
     *
     * @param originLocation   原文件: YYYYMMDD_01_AC_COMTRX.Z
     * @param destLocation     处理后文件: YYYYMMDD_01_AC_COMTRX_GW.zip
     * @param destZipInnerFile 处理后文件解压后文件名: YYYYMMDD_01_AC_COMTRX_GW
     * @return
     */
    private boolean convertFileFormat(String originLocation, String destLocation, String destZipInnerFile) {

        try {
            // Step 1，先删除YYYYMMDD_01_AC_COMTRX_GW.zip文件，防止已存在此文件导致写入失败
//            boolean delete = FileUtils.deleteFile(destLocation);
//            System.out.println("删除原文件：" + delete);
//            System.out.println();

            // Step 2，解压原文件: YYYYMMDD_01_AC_COMTRX.Z 为 tempRawFileName
//            String tempRawFileName = "/Users/vonmax/Downloads/merge/" + "tempRawBankFile.txt";
//            tempRawBankFile = unZFile(new File(originLocation), tempRawFileName);
//            if (null == tempRawBankFile) {
//                return false;
//            }

            // Step 3，拆分tempRawFileName 文件为多个_i 后缀的小文件，并删除大文件

            String tempRawFileName = "/Users/vonmax/Downloads/merge/tempRawBankFile.txt";

            splitFile(tempRawFileName, DEFAULT_SPLIT_FILE_SIZE);
//            new File(tempRawFileName).delete();

            // Step 4，创建与拆分子文件个数相同的线程分别处理子文件
            CountDownLatch latch = new CountDownLatch(DEFAULT_SPLIT_FILE_SIZE);
            for (int i = 1; i <= DEFAULT_SPLIT_FILE_SIZE; i++) {
                ParallelProcessBillService service = new ParallelProcessBillService(latch, i);//TODO:
                Runnable task = () -> service.exec();
                Thread thread = new Thread(task);
                thread.start();
            }

            System.out.println("start to process split bank raw file individually and await. ");
            latch.await();
            System.out.println("sub thread process split bank raw file success!, finish await");

            // Step 5，合并被处理的子文件为总文件tempRawProcessedBankFile.txt，并做压缩处理
            String finalTempFileName = "/Users/vonmax/Downloads/merge/tempRawProcessedBankFile.txt";
            mergeFile(finalTempFileName + "_", DEFAULT_SPLIT_FILE_SIZE, finalTempFileName);
            boolean compressFileResult = compressZipFile(new File(finalTempFileName), destLocation, destZipInnerFile);
            return compressFileResult;
        } catch (FileNotFoundException e) {
            System.out.println("文件不存在:");
        } catch (IOException e) {
            System.out.println("数据读取异常：");
        } catch (Exception e) {
            System.out.println("未知异常：");
        }
        return false;
    }

    private String handleBillLine(String line) {
        return line + " -> " + line;

    }

    /**
     * 并行处理账单拆分子文件的内部类
     */
    class ParallelProcessBillService {
        private CountDownLatch latch;
        private int processId;

        public ParallelProcessBillService(CountDownLatch latch, int processId) {
            this.latch = latch;
            this.processId = processId;
        }

        public void exec() {
            try {
                System.out.println("parallel start to process bank file: " + processId);
                parallelProcessFile(this.processId);
            } catch (Exception e) {
                System.out.println("parallel process bank file failure!：");
            } finally {
                latch.countDown();
            }
        }
    }

    /**
     * 并行处理文件，添加账单信息主方法
     *
     * @param processId
     * @return
     * @throws Exception
     */
    public boolean parallelProcessFile(int processId) throws Exception {
        // 创建临时处理文件
        File tempRawBankFile = new File("/Users/vonmax/Downloads/merge/" + "tempRawBankFile.txt_" + processId);
        File tempProcessedBankFile = new File("/Users/vonmax/Downloads/merge/" + "tempRawProcessedBankFile.txt_" + processId);

        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(tempRawBankFile), "utf-8"));
        BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(tempProcessedBankFile), "UTF-8"));

        String lineStr;
        int line = 0;
        String newLineStr;
        while ((lineStr = bufferedReader.readLine()) != null) {
            line++;
            if (line == 1 && processId == 1) {//如果是第一个文件的第一行，判断为标题行，需保留原样
                newLineStr = lineStr;
            } else {
                //处理每一条流水
                newLineStr = handleBillLine(lineStr);
            }

            if (StringUtils.isBlank(newLineStr)) {
                System.out.println("handle bill line error, error origin line:" + lineStr);
                return false;
            } else {
                bufferedWriter.write(newLineStr + "\n");
            }

        }

        bufferedWriter.flush();
        bufferedReader.close();
        bufferedWriter.close();

        // 删除临时原始子文件
//        tempRawBankFile.delete();//TODO
        return true;
    }

    /**
     * 拆分文件为小文件
     *
     * @param filePath
     * @param splitCount
     * @throws IOException
     */
    public static void splitFile(String filePath, int splitCount) throws IOException {
        FileInputStream fis = new FileInputStream(filePath);
        FileChannel inputChannel = fis.getChannel();
        final long fileSize = inputChannel.size();
        long average = fileSize / splitCount;//平均值
        long bufferSize = 200; //缓存块大小，自行调整
        ByteBuffer byteBuffer = ByteBuffer.allocate(Integer.valueOf(bufferSize + "")); // 申请一个缓存区
        long startPosition = 0; //子文件开始位置
        long endPosition = average < bufferSize ? 0 : average - bufferSize;//子文件结束位置
        for (int i = 0; i < splitCount; i++) {
            if (i + 1 != splitCount) {
                int read = inputChannel.read(byteBuffer, endPosition);// 读取数据
                readW:
                while (read != -1) {
                    byteBuffer.flip();//切换读模式
                    byte[] array = byteBuffer.array();
                    for (int j = 0; j < array.length; j++) {
                        byte b = array[j];
                        if (b == 10 || b == 13) { //判断\n\r
                            endPosition += j;
                            break readW;
                        }
                    }
                    endPosition += bufferSize;
                    byteBuffer.clear(); //重置缓存块指针
                    read = inputChannel.read(byteBuffer, endPosition);
                }
            } else {
                endPosition = fileSize; //最后一个文件直接指向文件末尾
            }

            FileOutputStream fos = new FileOutputStream(filePath + "_" + (i + 1));
            FileChannel outputChannel = fos.getChannel();
            inputChannel.transferTo(startPosition, endPosition - startPosition, outputChannel);//通道传输文件数据
            outputChannel.close();
            fos.close();
            startPosition = endPosition + 1;
            endPosition += average;
        }
        inputChannel.close();
        fis.close();
    }

    /**
     * 合并小文件为整体文件
     *
     * @param filePrefix
     * @param splitSize
     * @param finalFileName
     * @throws IOException
     */
    public static void mergeFile(String filePrefix, int splitSize, String finalFileName) throws Exception {
        File mergedFile = new File(finalFileName);
        FileInputStream in = null;
        FileChannel inChannel = null;

        FileOutputStream out = new FileOutputStream(mergedFile, true);
        FileChannel outChannel = out.getChannel();

        // 记录新文件最后一个数据的位置
        long start = 0;
        for (int i = 1; i <= splitSize; i++) {
            File file = new File(filePrefix + i);
            in = new FileInputStream(file);
            inChannel = in.getChannel();

            // 从inChannel中读取file.length()长度的数据，写入outChannel的start处
            outChannel.transferFrom(inChannel, start, file.length());
            start += file.length();
            in.close();
            inChannel.close();
//            file.delete();  //删除原始文件
        }

        out.close();
        outChannel.close();
    }


    public static File unZFile(File file, String outFileName) {
        int buffersize = DEFAULT_BUFFER_SIZE;
        FileOutputStream out = null;
        ZCompressorInputStream zIn = null;
        try {
            FileInputStream fin = new FileInputStream(file);
            BufferedInputStream in = new BufferedInputStream(fin);
            File outFile = new File(outFileName);
            out = new FileOutputStream(outFile);
            zIn = new ZCompressorInputStream(in);
            final byte[] buffer = new byte[buffersize];
            int n = 0;
            while (-1 != (n = zIn.read(buffer))) {
                out.write(buffer, 0, n);
            }
            return outFile;
        } catch (Exception e) {
            System.out.println("解压Z文件失败");
        } finally {
            try {
                if (out != null) {
                    out.close();
                }
                if (zIn != null) {
                    zIn.close();
                }
            } catch (IOException e) {
                System.out.println("解压Z文件关闭流失败");
            }
        }
        return null;
    }

    public static boolean compressZipFile(File rawFile, String outFileName, String outZipInnerName) {
        long startTime = System.currentTimeMillis();
        ZipOutputStream zos = null;
        InputStream is = null;
        try {
            zos = new ZipOutputStream(new FileOutputStream(outFileName));
            ZipEntry entry = new ZipEntry(outZipInnerName);//实例化压缩实体
            zos.setComment("处理后的银行账单压缩zip文件");  //设置注释
            zos.putNextEntry(entry);                    // 设置zip文件压缩实体, 也就是解压后的文件名
            is = new FileInputStream(rawFile);
            byte[] buffer = new byte[DEFAULT_BUFFER_SIZE];
            int readSize = 0;
            while (-1 != (readSize = is.read(buffer))) {
                zos.write(buffer, 0, readSize);
            }

            return true;
        } catch (Exception e) {
            System.out.println("压缩处理后的账单文件失败");
        } finally {
            try {
                if (null != is) {
                    is.close();
                }

                if (null != zos) {
                    zos.close();
                }

                rawFile.delete();   //删除原始文件
            } catch (Exception e) {
                System.out.println("压缩处理后的账单文件，关闭流失败");
            }

            long endTime = System.currentTimeMillis();
            System.out.println("compressZipFile in " + (endTime - startTime) + " milliseconds.");
        }

        return false;
    }


}
